package com.navi.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.navi.entity.ColumnEntity;
import com.navi.entity.TableEntity;
import com.navi.exception.HmsException;
import com.navi.pojo.KeyValueForTimestampPojo;
import com.navi.util.LogUtil;
import com.navi.util.SplitTimestampUtil;
import com.navi.util.TableJdbcUtil;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.sql.*;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.Date;
import java.util.stream.Collectors;

import static com.navi.constant.GenericDef.*;
import static com.navi.constant.GenericDef.WPP_SHT_OPE_HIS_TABLE;

/**
 * Created by ghost on 2018/11/19.
 */
@Service
public class StageTableOperationService {

    public static final String MK_MAX_TIME = "maxt";
    public static final String MK_MIN_TIME = "mint";
    public static final String MK_OPES = "opes";

    @Value("${stagetable.schema}")
    private String stageTableSchema;
    @Value("${external.table.schema}")
    private String extTableSchema;
    @Value("${gpfdist.host}")
    private String gpfdistHost;
    @Value("${gpfdist.port}")
    private String gpfdistPort;
    @Value("${app.conf.table.sht_ope_his}")
    private String shtOpeHisTables;

    @Autowired
    private TableMetaService tableMetaService;

    @Autowired
    private ExtTableOperationService extTableOperationService;

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    /*
    创建临时表
     */
    public String createStageTable(Statement stmt, String stageTableName, TableEntity tableEntity) throws SQLException {
        /**
         * find stage table can reuse
         */
        logger.info("get stageTableSchema [{}] info ... ", stageTableName);
        final TableEntity stageTableEntity = tableMetaService.getTableMetaInfo(stageTableSchema, stageTableName, stmt);
        logger.info("comparing table meta [{}] ... ", stageTableName);
        //已存在的stage table和即将要创建的stage table一样时，直接使用，不再重新创建
        boolean isSame = stageTableEntity != null && extTableOperationService.compare(stageTableEntity, tableEntity);

        if (isSame) {
            logger.info("{} : reused stage table ", stageTableName);
            long endTime = System.currentTimeMillis();
            String sql = String.format("truncate %s.%s", stageTableSchema, stageTableName);
            stmt.execute(sql);
            logger.info("{}, cost {} ms", sql, System.currentTimeMillis() - endTime);

            /*** 田老师增加逻辑: defect partition表 0-5分才做VACUUM FULL **/
            SimpleDateFormat df = new SimpleDateFormat("mm");
            int minutes = Integer.parseInt(df.format(new Date()));
            logger.info("35分-37分做 defect_f_n VACUUM FULL,40分-42分做 pds VACUUM FULL,45分-47分做 mds VACUUM FULL,50分-52分做 ope_his VACUUM FULL,55分-57分做 chamber_hst VACUUM FULL,每次都做ANALYZE，系统当前分钟: " + minutes);
            if (
                    (stageTableName.contains("defect_f_n"))&&(minutes >= 35 && minutes <= 37)||
                            (stageTableName.contains("pds"))&&(minutes >= 40 && minutes <= 42)||
                            (stageTableName.contains("mds"))&&(minutes >= 45 && minutes <= 47)||
                            (stageTableName.contains("sht_ope_his"))&&(minutes >= 50 && minutes <= 52)||
                            (stageTableName.contains("chamber_hst"))&&(minutes >= 55 && minutes <= 57)
                    ) {
                endTime = System.currentTimeMillis();
                String sqlVacuum = String.format("VACUUM FULL %s.%s", stageTableSchema, stageTableName);
                stmt.execute(sqlVacuum);
                logger.info("{}, cost {} ms", sqlVacuum, System.currentTimeMillis() - endTime);
            }
            endTime = System.currentTimeMillis();
            String sqlVacuum2 = String.format("ANALYZE %s.%s", stageTableSchema, stageTableName);
            stmt.execute(sqlVacuum2);
            logger.info("{}, cost {} ms", sqlVacuum2, System.currentTimeMillis() - endTime);


        } else {
            //drop table if exists different stage table
            String sql = String.format("drop table if exists %s.%s", stageTableSchema, stageTableName);
            stmt.execute(sql);

            //build create table ddl
            String beingCreatedSql = buildDll(stageTableSchema, tableEntity, stageTableName);
            logger.info("created stage table ddl : {}", beingCreatedSql);

            //run ddl
            stmt.execute(beingCreatedSql);

            //update cache
            tableMetaService.saveTableMetaInfo2Map(stageTableSchema, stageTableName, stmt);
        }

        return stageTableName;
    }

    private String buildDll(String tableSchema, TableEntity tableEntity, String stageTableName) {
        String sql;
        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();

        String columnString = columnEntityList.stream()
                .map(cl -> cl.getColumnName() + " " + cl.getDataType())
                .collect(Collectors.joining(","));

        List<ColumnEntity> primaryColumnList = columnEntityList.stream()
                .filter(ColumnEntity::getPrimaryKeyFlg)
                .collect(Collectors.toList());


        //create stage table
        //build ddl for create table
        //has no primary key in gpload.py , so ...
        StringBuffer sqlBuffer = new StringBuffer();
        sqlBuffer.append("CREATE TABLE ").append(tableSchema).append(".").append(stageTableName);
        sqlBuffer.append("(").append(columnString).append(")");

        logger.info("创建stage表时更改表空间语句: TABLESPACE data01tbs");
        sqlBuffer.append(" TABLESPACE data01tbs ");

        if (!primaryColumnList.isEmpty()) {
            final String distCols = primaryColumnList.stream()
                    .map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
            sqlBuffer.append(" DISTRIBUTED BY (").append(distCols).append(")");
        }

        sql = sqlBuffer.toString();
        return sql;
    }



    /**
     * 查询stage表中的站点分布、时间范围，供后续delete使用
     * 获取站点列表和最大时间&最小时间，优化 partition 影响
     * 主要针对defect相关表 有evt_Timestamp & ope_id
     * 增加TEG的 proc_end & ope_id
     *
     * @param statement
     * @param tableName
     * @param stageTableSchema
     * @param stageTableName
     * @return
     * @throws SQLException
     */
    public Map<String, String> collectStageTableDataInfo(Statement statement, String tableName, String stageTableSchema, String stageTableName) throws SQLException {
        Map<String, String> stageInfoMap = new HashMap<>();

        String timeColumn = "evt_Timestamp";
        if (tableName.startsWith(EDC_DATA_TABLE_PREFIX)) {
            timeColumn = "event_time";
        }

        //get detail condition
        String sqlStg = "SELECT min(" + timeColumn + ") mint ,max(" + timeColumn + ") maxt," +
                "ARRAY_TO_STRING(ARRAY_AGG(DISTINCT ope_id),',') as opes FROM " + stageTableSchema + "." + stageTableName;

        final ResultSet rsStg = statement.executeQuery(sqlStg);

        String mint = null, maxt = null, opes = null, opeCondition = null;
        try {
            while (rsStg.next()) {
                mint = rsStg.getString("mint");
                maxt = rsStg.getString("maxt");
                opes = rsStg.getString("opes");
            }
        }catch (SQLException e){
            throw e;
        }finally {
            rsStg.close();
        }
        if (StringUtils.isEmpty(opes) || StringUtils.isEmpty(mint) || StringUtils.isEmpty(maxt)) {
            logger.error("stage表过滤条件有空！stageTable {}.{}, ope {}, maxEvtTime {}, minEvtTime {}",
                    stageTableSchema, stageTableName, opes, maxt, mint);
            throw new HmsException("在 collectStageTableDataInfo 函数中获取stage表的站点和时间错误");
        }
        opes = opes.trim();
        logger.info("collectStageTableDataInfo : evt Time:[{} ~ {}], opes:{}.", mint, maxt, opes);
        opeCondition = Arrays.stream(opes.split(","))
                .map(o -> "'" + o + "'").collect(Collectors.joining(","));

        stageInfoMap.put(MK_MAX_TIME, maxt);
        stageInfoMap.put(MK_MIN_TIME, mint);
        stageInfoMap.put(MK_OPES, opeCondition);

        return stageInfoMap;
    }


    /*** 为eda时查询stage表中的机台分布、时间范围，供后续delete使用 **/
    public Map<String, String> edaCollectStageTableDataInfo(Statement statement, String tableName, String stageTableSchema, String stageTableName) throws SQLException {
        Map<String, String> stageInfoMap = new HashMap<>();


        String timeColumn = "";
        String opeColumn = "equip_id";
        if(tableName.contains(ARRAY_PDS_GLASS_SUMMARY)||tableName.contains(CF_PDS_GLASS_SUMMARY)) {
            timeColumn="glass_start_time";
        }else if(tableName.contains(CELL_PDS_COMPT_SUMMARY)) {
            timeColumn="component_start_time";
        } else if(tableName.startsWith(CELL2_PDS_CHIP_SUMMARY)) {
            timeColumn="chip_start_time";
        }else if(tableName.contains(ARRAY_MDS_GLASS_SUMMARY)||tableName.contains(CF_MDS_GLASS_SUMMARY)) {
            timeColumn="glass_start_time";
        }else if(tableName.contains(CELL_MDS_COMPT_SUMMARY)) {
            timeColumn="component_start_time";
        } else if(tableName.startsWith(CELL2_MDS_CHIP_SUMMARY)) {
            timeColumn="chip_start_time";
        }
        else {
            timeColumn="glass_start_time";
        }

        //get detail condition
        //String sqlStg = "SELECT min(substring(glass_id,3,2)) mint ,max(" + timeColumn + ") maxt," +
        //        "ARRAY_TO_STRING(ARRAY_AGG(DISTINCT " + opeColumn + "),',') as opes FROM " + stageTableSchema + "." + stageTableName;
        String sqlStg = "SELECT min(" + timeColumn + ") mint ,max(" + timeColumn + ") maxt," +
                "ARRAY_TO_STRING(ARRAY_AGG(DISTINCT " + opeColumn + "),',') as opes FROM " + stageTableSchema + "." + stageTableName;

        final ResultSet rsStg = statement.executeQuery(sqlStg);

        String mint = null, maxt = null, opes = null, opeCondition = null;
        try {
            while (rsStg.next()) {
                //String minGlassTime = rsStg.getString("mint");
                //String year = minGlassTime.substring(0, 1);
                //String month = minGlassTime.substring(1, 2);
                //if (month.equalsIgnoreCase("A")) {
                //    month = "10";
                //}else if (month.equalsIgnoreCase("B")) {
                //    month = "11";
                //}else if (month.equalsIgnoreCase("C")) {
                //    month = "12";
                //}else {
                //    month = "0" + month;
                //}
                //mint = "201" + year + "-" + month + "-01 00:00:00";
                mint = rsStg.getString("mint");
                maxt = rsStg.getString("maxt");
                opes = rsStg.getString("opes");
            }
        }catch (SQLException e){
            throw e;
        }finally {
            rsStg.close();
        }
        if (StringUtils.isEmpty(opes) || StringUtils.isEmpty(mint) || StringUtils.isEmpty(maxt)) {
            logger.error("stage表过滤条件有空！stageTable {}.{}, ope {}, maxEvtTime {}, minEvtTime {}",
                    stageTableSchema, stageTableName, opes, maxt, mint);
            throw new HmsException("在 collectStageTableDataInfo 函数中获取stage表的站点和时间错误\n"+sqlStg);
        }
        opes = opes.trim();
        logger.info("collectStageTableDataInfo : evt Time:[{} ~ {}], opes:{}.", mint, maxt, opes);
        opeCondition = Arrays.stream(opes.split(","))
                .map(o -> "'" + o + "'").collect(Collectors.joining(","));

        stageInfoMap.put(MK_MAX_TIME, maxt);
        stageInfoMap.put(MK_MIN_TIME, mint);
        stageInfoMap.put(MK_OPES, opeCondition);

        return stageInfoMap;
    }





    /**
     * 从目标表插入数据到stage表
     * 更新stage表
     * 删除目标表
     * 从stage表导入数据到目标表
     *
     * @param stmt
     * @param targetTableSchema
     * @param targetTableName
     * @param stageTableName
     * @param tableEntity
     * @param jsonObjectList
     * @return
     * @throws SQLException
     */
    public int updateTargetTableByStageTable(Jedis jedis, Connection connection, Statement stmt, String targetTableSchema, String targetTableName, String stageTableName,
                                              TableEntity tableEntity, List<JSONObject> jsonObjectList) throws SQLException {

//        QueryRunner qr = new QueryRunner();
        long lBeginTime = System.currentTimeMillis();
        if (jsonObjectList.isEmpty()) {
            return 0;
        }
        int jsonObjectCnt = jsonObjectList.size();
        logger.info("InsertStageTableFromTargetTable total {} sql", jsonObjectCnt);

        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final List<String> columnNameList = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.toList());
        final String columnString = columnNameList.stream().collect(Collectors.joining(","));
        final List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        final String matchString = primaryColumnList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));

        // insert
        //String querysql = jsonObjectList.stream()
        //        .filter(jo -> jo.containsKey("query_conditions") && jo.containsKey("update_columns"))
        //        .map(jo -> (JSONObject) jo.get("query_conditions"))
        //        .map(qc ->
        //                qc.keySet().stream().map(k -> k + "='" + qc.get(k) + "'")
        //                        .collect(Collectors.collectingAndThen(Collectors.joining(" AND "), x -> "(" + x + ")"))
        //        ).collect(Collectors.joining(" OR "));
        /*********** 8.7修改 将 or 语句修改为 in *********/
        List<String> querysqlList = new ArrayList<>();
        List<String> keyStringList = new ArrayList<>(
                jsonObjectList.stream()
                        .filter(jo -> jo.containsKey("query_conditions") && jo.containsKey("update_columns"))
                        .map(jo -> (JSONObject) jo.get("query_conditions"))
                        .map(jo -> {
                            List<String> keyList = new ArrayList<>(jo.keySet());
                            Collections.sort(keyList);
                            return keyList.stream().collect(Collectors.joining(","));
                        }).collect(Collectors.toSet())
        );
        keyStringList.forEach(keyString -> {
            String querysql = jsonObjectList.stream()
                    .filter(jo -> jo.containsKey("query_conditions") && jo.containsKey("update_columns"))
                    .map(jo -> (JSONObject) jo.get("query_conditions"))
                    .filter(qc -> {
                        List<String> keyList = new ArrayList<>(qc.keySet());
                        Collections.sort(keyList);
                        return keyString.equals(keyList.stream().collect(Collectors.joining(",")));
                    })
                    .map(qc -> {
                        return Arrays.stream(keyString.split(","))
                                .map(k -> "'" + qc.get(k) + "'")
                                .collect(Collectors.joining(",", "(", ")"));
                    }).collect(Collectors.joining(" , "));
            querysql = "(" + keyString + ") in (" + querysql + ")";
            querysqlList.add(querysql);
        });

        StringBuffer sqlBuffer = new StringBuffer("insert into ");
        sqlBuffer.append(stageTableSchema).append(".").append(stageTableName).append("(").append(columnString).append(")");
        sqlBuffer.append("( select ").append(columnString).append(" from ").append(targetTableSchema).append(".").append(targetTableName)
                .append(" where ");

        int effectRows = 0;

        for (int i = 0; i < querysqlList.size(); i++) {
            final String sql = sqlBuffer.toString() + querysqlList.get(i) + ")";
            logger.info("****** 数据库层面提升性能语句 ****** Sql: set optimizer='off'");
            stmt.execute("set optimizer='off'");

            logger.info("insert stage sql : {}", sql);
            int effectRow = stmt.executeUpdate(sql);
            effectRows += effectRow;
            logger.info("executeBatch insert {} rows into stage table {} costs {} millis", effectRow, stageTableName, (System.currentTimeMillis() - lBeginTime));
            lBeginTime = LogUtil.printPerformance(LOG_UPDATE_REAL_FLG_DELETE_INSERT_INSERT_INTO_STAGR, stageTableName, effectRow, lBeginTime);
        }

        if (effectRows <= 0) {
            for (int i = 0; i < querysqlList.size(); i++) {
                String sql = sqlBuffer.toString() + querysqlList.get(i) + ")";
                logger.warn("update Q find {} rows from {}, sql is {}", effectRows, targetTableSchema, sql);
            }
            logger.warn("update Q {}.{}, {}", targetTableSchema, targetTableName, JSON.toJSONString(jsonObjectList));
            return effectRows;
        }

        // update stage
        sqlBuffer = new StringBuffer("update ");
        sqlBuffer.append(stageTableSchema).append(".").append(stageTableName).append(" set ");
        String updateStage = sqlBuffer.toString();
        List<String> updateStageList = new ArrayList<>();
        updateStageList = jsonObjectList.stream()
                .map(jo -> {
                            JSONObject uc = (JSONObject) jo.get("update_columns");
                            JSONObject qc = (JSONObject) jo.get("query_conditions");
                            return updateStage +
                                    uc.keySet().stream().map(k -> k + "='" + uc.get(k) + "'").collect(Collectors.joining(" , "))
                                    + " where "
                                    + qc.keySet().stream().map(k -> k + "='" + qc.get(k) + "'").collect(Collectors.joining(" AND "));
                        }
                ).collect(Collectors.toList());

        int totalEffectRows = 0;
        long lBeginTimeTmp;
        for (int i = 0; i < updateStageList.size(); i++) {
            logger.info("update stage sql : {}", updateStageList.get(i));
            lBeginTimeTmp = System.currentTimeMillis();
            effectRows = stmt.executeUpdate(updateStageList.get(i));
            totalEffectRows += effectRows;
            logger.info("update stage {} rows in stage table {} costs {} millis",
                    effectRows, stageTableName, (System.currentTimeMillis() - lBeginTimeTmp));
        }
        lBeginTime = LogUtil.printPerformance(LOG_UPDATE_REAL_FLG_DELETE_INSERT_UPDATE_STAGE, stageTableName, totalEffectRows, lBeginTime);

        String timeColumn = "evt_Timestamp";

        //get detail condition
        String sqlStg = "SELECT min(" + timeColumn + ") mint ,max(" + timeColumn + ") maxt," +
                "ARRAY_TO_STRING(ARRAY_AGG(DISTINCT ope_id),',') as opes FROM " + stageTableSchema + "." + stageTableName;
        final ResultSet rsStg = stmt.executeQuery(sqlStg);

        String mint = null, maxt = null, opes = null, opeCondition = null;
        try {
            while (rsStg.next()) {
                mint = rsStg.getString("mint");
                maxt = rsStg.getString("maxt");
                opes = rsStg.getString("opes");
            }
        }catch (SQLException ex){
            throw ex;
        }finally {
            rsStg.close();
        }
        if (StringUtils.isEmpty(opes) || StringUtils.isEmpty(mint) || StringUtils.isEmpty(maxt)) {
            throw new HmsException("在updateTargetTableByStageTable函数中获取stage表的站点和时间错误");
        }
        opes = opes.trim();
        logger.info("stage表的站点和时间: evt Time:[{} ~ {}], opes:{}.", mint, maxt, opes);
        opeCondition = Arrays.stream(opes.split(","))
                .map(o -> "'" + o + "'").collect(Collectors.joining(","));

        // delete target table
        sqlBuffer = new StringBuffer("delete from ");
        sqlBuffer.append(targetTableSchema).append(".").append(targetTableName).append(" del_table using ")
                .append(stageTableSchema).append(".").append(stageTableName).append(" stg_table ");

        final String onConditionString = primaryColumnList.stream()
                .map(ColumnEntity::getColumnName)
                .map(columnName -> String.format(" del_table.%s = stg_table.%s ", columnName, columnName))
                .collect(Collectors.joining(" and "));

        sqlBuffer.append(" where ").append(onConditionString)
                //.append(" and del_table.").append(timeColumn).append(" between '")
                //.append(mint).append("' and '").append(maxt).append("' ")
                .append(" and del_table.").append("ope_id in (").append(opeCondition).append(")");

        int totalDelEffectRow = 0;

        List<KeyValueForTimestampPojo> KeyValueForTimestampPojoList = SplitTimestampUtil.getKeyValueForDate(mint, maxt);
        for (KeyValueForTimestampPojo pojo : KeyValueForTimestampPojoList) {
            long beginTime = System.currentTimeMillis();
            String byTimePartitionSql = " and del_table." + timeColumn + " between '" + pojo.getStartTiemstamp().toString() + "' and '" + pojo.getEndTimestamp().toString() + "'";
            logger.info("delete by time partition {} sql : {}", targetTableName, sqlBuffer.toString() + byTimePartitionSql);
            int effectRow = 0;
            effectRows = stmt.executeUpdate(sqlBuffer.toString() + byTimePartitionSql);
            totalDelEffectRow += effectRows;
            logger.info("delete by time partition {} rows from table {} costs {} millis", effectRows, targetTableName, (System.currentTimeMillis() - beginTime));
        }

        //logger.info("delete {} sql : {}", targetTableName, sqlBuffer.toString());
        //effectRows = stmt.executeUpdate(sqlBuffer.toString());
        logger.info("delete total {} rows from table {} costs {} millis", totalDelEffectRow, targetTableName, (System.currentTimeMillis() - lBeginTime));
        lBeginTime = LogUtil.printPerformance(LOG_UPDATE_REAL_FLG_DELETE_INSERT_DELETE_TARGET_TABLE, targetTableName, effectRows, lBeginTime);

        // insert into target table
        sqlBuffer = new StringBuffer("insert into ");
        sqlBuffer.append(targetTableSchema).append(".").append(targetTableName);
        sqlBuffer.append("(").append(columnString).append(")");

        final String fromColumnString = columnEntityList.stream()
                .map(ColumnEntity::getColumnName)
                .map(columnName -> String.format("from_table.%s", columnName))
                .collect(Collectors.joining(","));

        // 从stage表插入目标表的时候需要用over partition by， 因为stage表中可能会有重复的数据
        sqlBuffer.append(" select ").append(fromColumnString).append(" from ")
                .append(" (select *, row_number() over (partition by ").append(matchString).append(")").append(" as gpload_row_number ")
                .append(" from ").append(stageTableSchema).append(".").append(stageTableName).append(")").append(" as from_table")
                .append(" where gpload_row_number = 1 ");

        String insertSql = sqlBuffer.toString();
        logger.info("do insert , sql: {}", insertSql);
        lBeginTime = System.currentTimeMillis();
        final int insertRows = stmt.executeUpdate(insertSql);
        logger.info("insert records rows {} to table {} costs {} millis",
                insertRows, targetTableName, (System.currentTimeMillis() - lBeginTime));
        lBeginTime = LogUtil.printPerformance(LOG_UPDATE_REAL_FLG_DELETE_INSERT_INSERT_INTO_TARGET_TABLE, targetTableName, effectRows, lBeginTime);
        // todo 这部分代码需要抽离出通用逻辑

        return effectRows;
    }


    /**
     * 数据从EXT表插入stage表
     *
     * @param stmt
     * @param extTableName
     * @param stageTableName
     * @param tableEntity
     * @param jsonObjectList
     * @return
     * @throws SQLException
     */
    public int doInsertStageTable(Statement stmt, String extTableName, String stageTableName, TableEntity tableEntity, List<JSONObject> jsonObjectList) throws SQLException {
        long lBeginTime = System.currentTimeMillis();
        if (jsonObjectList.isEmpty()) {
            return 0;
        }

        int jsonObjectCnt = jsonObjectList.size();
        logger.info("total {} rows", jsonObjectCnt);

        // over (partition by
        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final List<String> columnNameList = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.toList());
        final String columnString = columnNameList.stream().collect(Collectors.joining(","));
        List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        //real_flg 去重逻辑
        boolean oldOpeHis=Arrays.asList(shtOpeHisTables.split(",")).stream().filter(element -> stageTableName.contains(element)).count() > 0;
        boolean newOpeHis=Arrays.asList(WPP_SHT_OPE_HIS_TABLE.split(",")).stream().filter(element -> stageTableName.contains(element)).count() > 0;
        boolean mdsGlass=Arrays.asList(MDS_GLASS_TABLE.split(",")).stream().filter(element -> stageTableName.contains(element)&&!stageTableName.contains("summary")).count() > 0;
        boolean chamberHst=Arrays.asList(CHAMBER_HST_TABLE.split(",")).stream().filter(element -> stageTableName.contains(element)).count() > 0;

        if (oldOpeHis||newOpeHis)   {
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"logof_timestamp".equals(entity.getColumnName()))
                    .filter(entity -> !"logofTimestamp".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }else if (chamberHst) {
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }else if(mdsGlass){
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"equip_id".equals(entity.getColumnName()))
                    .filter(entity -> !"equipId".equals(entity.getColumnName()))
                    .filter(entity -> !"sub_equip_id".equals(entity.getColumnName()))
                    .filter(entity -> !"subEquipId".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }



        final String matchString = primaryColumnList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));

        final String fromColumnString = TableJdbcUtil.getColumnStringWithTableAlias(columnEntityList, "from_table");

        // insert
        StringBuilder insertSqlBuilder = new StringBuilder("insert into ");
        insertSqlBuilder.append(stageTableSchema).append(".").append(stageTableName).append("(").append(columnString).append(")");

        // 从 ext 表插入 stage 表的时候需要用over partition by， 因为 ext 表中可能会有重复的数据
        // 特殊逻辑： sht_ope表去重时保留最新的 logof_timestamp 数据  //10张量测机台过账记录表 主键一样时 保留最后一次的解析记录
        String partitionOrderCondition = "";
        if (oldOpeHis||newOpeHis) {
            partitionOrderCondition = " order by logof_timestamp desc ";
        }else if(mdsGlass||chamberHst){
            if(stageTableName.contains("array_")||stageTableName.contains("cf_")) {
                partitionOrderCondition = " order by glass_start_time desc ";
            }else if(stageTableName.contains("cell_")){
                partitionOrderCondition = " order by component_start_time desc ";
            }else if(stageTableName.contains("cell2_")){
                partitionOrderCondition = " order by chip_start_time desc ";
            }
        }
//        else if(stageTableName.contains("wpp_sht_ope_his_ext")){
//            partitionOrderCondition = " order by etl_timestamp desc ";
//        }

        if( newOpeHis||mdsGlass||chamberHst){
            insertSqlBuilder.append(" select distinct ").append(fromColumnString).append(" from ").append(extTableSchema)
                    .append(".").append(extTableName) .append(" as from_table ");
        }else{
            insertSqlBuilder.append(" select ").append(fromColumnString).append(" from ")
                    .append(" (select *, row_number() over (partition by ").append(matchString).append(partitionOrderCondition).append(")").append(" as gpload_row_number ")
                    .append(" from ").append(extTableSchema).append(".").append(extTableName).append(")").append(" as from_table")
                    .append(" where gpload_row_number = 1 ");
        }
        final String sql = insertSqlBuilder.toString();
        logger.info("insert stage sql : {}", sql);
        int effectRows = stmt.executeUpdate(sql);
        logger.info("executeBatch insert {} rows into stage table {} costs {} millis", effectRows, stageTableName, (System.currentTimeMillis() - lBeginTime));

        return effectRows;
    }

    public int doUpdate(Statement statement, String fromTableName, String intoTableSchema, String intoTableName, TableEntity tableEntity, String limit_time) throws SQLException {

        Timestamp currenTime = Timestamp.valueOf(LocalDateTime.now());

        StringBuffer sqlBuffer = new StringBuffer("update ");
        sqlBuffer.append(intoTableSchema).append(".").append(intoTableName).append(" into_table");
        sqlBuffer.append(" set ");

        final String updateColumnString = tableEntity.getColumnEntityList().stream()
                .filter(columnEntity -> !columnEntity.getPrimaryKeyFlg())
                .map(ColumnEntity::getColumnName)
                .map(column -> String.format(" %s = from_table.%s ", column, column))
                .collect(Collectors.joining(","));

        sqlBuffer.append(updateColumnString);

        sqlBuffer.append(" from ").append(stageTableSchema).append(".").append(fromTableName).append(" from_table");

        final String whereCondition = tableEntity.getColumnEntityList().stream()
                .filter(ColumnEntity::getPrimaryKeyFlg)
                .map(ColumnEntity::getColumnName)
                .map(column -> String.format(" into_table.%s = from_table.%s ", column, column))
                .collect(Collectors.joining("and"));

        sqlBuffer.append(" where").append(whereCondition);
        if ("true".equals(limit_time)) {
            sqlBuffer.append(" and into_table.evt_timestamp <= ").append("'").append(currenTime).append("'");
        }


        String sql = sqlBuffer.toString();
        logger.info("update sql : {}", sql);
        final int effectRows = statement.executeUpdate(sql);

        logger.info("update effected rows {}", effectRows);
        return effectRows;
    }


    /**
     * stage insert into target table
     * 全表插入，调用之前先删除目标表中主键相同的记录
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param columnString
     * @return
     * @throws SQLException
     */
    public int insertStage2TargetTable(Statement statement,
                                        String stageTableSchema, String stageTableName,
                                        String tableSchema, String tableName,
                                        String columnString) throws SQLException {
        long beginTime = System.currentTimeMillis();
        String sql = new StringBuilder().append("insert into ")
                .append(tableSchema).append(".").append(tableName)
                .append("(").append(columnString).append(")")
                .append("( select ").append(columnString).append(" from ").append(stageTableSchema).append(".").append(stageTableName).append(")").toString();

        logger.info("do insert from stage, sql: {}.", sql);

        final int insertRows = statement.executeUpdate(sql);
        logger.info("insert {} rows from stage.", insertRows);
        LogUtil.printPerformance(LOG_INSERT_STAGE_2_TARGET_TABLE, tableName, insertRows, beginTime);
        LogUtil.printCost(LOG_INSERT_STAGE_2_TARGET_TABLE, tableName, "insert records rows [" + insertRows + "]", beginTime);

        return insertRows;
    }

    /**
     * 将stage表中数据插入目标表
     * 只包含主键不重复的数据
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param stageInfoMap      stage表的汇总信息
     * @param columnString      逗号分隔的列名
     * @param primaryColumnList
     * @param tableEntity
     * @return
     * @throws SQLException
     */
    private int insertNotExistRecordFromStage2TargetTable(Statement statement,
                                                          String stageTableSchema, String stageTableName,
                                                          String tableSchema, String tableName,
                                                          Map<String, String> stageInfoMap,
                                                          String columnString, List<ColumnEntity> primaryColumnList, TableEntity tableEntity) throws SQLException {
        long beginTime = System.currentTimeMillis();

        String sqlWithPartionCondition =
                " and into_table." + "evt_timestamp between '" +
                        stageInfoMap.get(MK_MIN_TIME) + "' and '" + stageInfoMap.get(MK_MAX_TIME) + "' " +
                        " and into_table." + "ope_id in (" + stageInfoMap.get(MK_OPES) + ") ";

        String sql = TableJdbcUtil.buildInsertNotExistedSql(stageTableSchema, stageTableName, tableSchema, tableName,
                "from_table", "into_table", columnString, primaryColumnList, tableEntity, sqlWithPartionCondition);

        logger.info("do insert from stage, sql: {}.", sql);
        final int insertRows = statement.executeUpdate(sql);
        logger.info("insert {} rows from stage.", insertRows);
        LogUtil.printPerformance(LOG_MERGE_PROC_DELETE_INSERT_I, tableName, insertRows, beginTime);
        LogUtil.printCost(LOG_MERGE_PROC_DELETE_INSERT_I, tableName, "insert records rows [" + insertRows + "]", beginTime);
        return insertRows;
    }

    /**
     * delete from target table by stage
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param primaryColumnList
     * @return
     * @throws SQLException
     */
    private int deleteTargetTableByStage(Statement statement,
                                         String stageTableSchema, String stageTableName,
                                         String tableSchema, String tableName,
                                         Map<String, String> stageInfoMap,
                                         List<ColumnEntity> primaryColumnList) throws SQLException {
        long beginTime = System.currentTimeMillis();
        int deleteRows = 0;

        String timeColumn = "evt_Timestamp";
        if (tableName.startsWith(EDC_DATA_TABLE_PREFIX)) {
            timeColumn = "event_time";
        }

        //delete exist
        StringBuilder delSql = new StringBuilder("delete from ");
        delSql.append(tableSchema).append(".").append(tableName).append(" del_table using ")
                .append(stageTableSchema).append(".").append(stageTableName).append(" stg_table ");
        final String onConditionString = TableJdbcUtil.getOnConditionStringByPk(primaryColumnList, "del_table", "stg_table");

        delSql.append(" where ").append(onConditionString)
                //.append(" and del_table.").append("evt_timestamp between '")
                //.append(stageInfoMap.get(MK_MIN_TIME)).append("' and '").append(stageInfoMap.get(MK_MAX_TIME)).append("' ")
                .append(" and del_table.").append("ope_id in (").append(stageInfoMap.get(MK_OPES)).append(")");

        List<KeyValueForTimestampPojo> KeyValueForTimestampPojoList = SplitTimestampUtil.getKeyValueForDate(stageInfoMap.get(MK_MIN_TIME), stageInfoMap.get(MK_MAX_TIME));
        for (KeyValueForTimestampPojo pojo : KeyValueForTimestampPojoList) {
            long beginTime2 = System.currentTimeMillis();
            String byTimePartitionSql = " and del_table." + timeColumn + " between '" + pojo.getStartTiemstamp().toString() + "' and '" + pojo.getEndTimestamp().toString() + "'";
            logger.info("do delete by stage by time partition, sql : {}", delSql + byTimePartitionSql);
            int deleteRow = statement.executeUpdate(delSql + byTimePartitionSql);
            deleteRows += deleteRow;
            logger.info("delete total {} rows by stage by time partition, costs {} millis", deleteRow, (System.currentTimeMillis() - beginTime2));
        }

        //String sql = delSql.toString();
        //logger.info("do delete by stage, sql: {}", sql);
        //final int deleteRows = statement.executeUpdate(sql);
        logger.info("delete total {} rows by stage", deleteRows);
        LogUtil.printPerformance(LOG_MERGE_PROC_DELETE_INSERT_D, tableName, deleteRows, beginTime);
        LogUtil.printCost(LOG_MERGE_PROC_DELETE_INSERT_D, tableName, "delete records rows [" + deleteRows + "]", beginTime);

        return deleteRows;
    }

    /**
     * delete from target table by stage
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param primaryColumnList
     * @return
     * @throws SQLException
     */
    public int deleteEdaTargetTableByStage(Statement statement,
                                            String stageTableSchema, String stageTableName,
                                            String tableSchema, String tableName,
                                            Map<String, String> stageInfoMap,
                                            List<ColumnEntity> primaryColumnList) throws SQLException {
        int deleteRows = 0;

        final String onConditionString = TableJdbcUtil.getOnConditionStringByPk(primaryColumnList, "del_table", "stg_table");

        //delete exist
        String delSorSql = "delete from " + tableSchema + "." + tableName + " del_table using " +
                stageTableSchema + "." + stageTableName + " stg_table " +
                " where " + onConditionString + " and del_table.equip_id in (" + stageInfoMap.get(MK_OPES) + ")";

        List<KeyValueForTimestampPojo> KeyValueForTimestampPojoList = SplitTimestampUtil.getKeyValueForDate(stageInfoMap.get(MK_MIN_TIME), stageInfoMap.get(MK_MAX_TIME));

        for (KeyValueForTimestampPojo pojo : KeyValueForTimestampPojoList) {
            long beginTime2 = System.currentTimeMillis();
            String byTimePartitionSql ="";
            if(tableName.contains(ARRAY_PDS_GLASS_SUMMARY)||tableName.contains(CF_PDS_GLASS_SUMMARY)||tableName.contains(ARRAY_MDS_GLASS_SUMMARY)||tableName.contains(CF_MDS_GLASS_SUMMARY)) {
                byTimePartitionSql=" and del_table.glass_start_time between '" + pojo.getStartTiemstamp().toString() + "' and '" + pojo.getEndTimestamp().toString() + "'";
            }else if(tableName.contains(CELL_PDS_COMPT_SUMMARY)||tableName.contains(CELL_MDS_COMPT_SUMMARY)) {
                byTimePartitionSql=" and del_table.component_start_time between '" + pojo.getStartTiemstamp().toString() + "' and '" + pojo.getEndTimestamp().toString() + "'";
            } else if(tableName.startsWith(CELL2_PDS_CHIP_SUMMARY)||tableName.startsWith(CELL2_MDS_CHIP_SUMMARY)) {
                byTimePartitionSql=" and del_table.chip_start_time between '" + pojo.getStartTiemstamp().toString() + "' and '" + pojo.getEndTimestamp().toString() + "'";
            } else{
                byTimePartitionSql=" and del_table.glass_start_time between '" + pojo.getStartTiemstamp().toString() + "' and '" + pojo.getEndTimestamp().toString() + "'";
            }
            logger.info("do delete by stage by time partition, sql : {}", delSorSql + byTimePartitionSql);
            int deleteRow = statement.executeUpdate(delSorSql + byTimePartitionSql);
            deleteRows += deleteRow;
            logger.info("delete {} rows by stage by time partition, costs {} millis", deleteRow, (System.currentTimeMillis() - beginTime2));
        }
        logger.info("delete records from sor table total rows {}", deleteRows);
        return deleteRows;
    }

    /**
     * 删除存在数据，stage数据全部insert -> 先 insert， 如果 insert 的数量和 stage 表数量不同（有主键重复的数据）， 则执行 delete & re insert
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param tableEntity
     * @param stageTableRows
     * @return
     * @throws SQLException
     */
    private int[] stageData2TargetTable(Statement statement,
                                        String stageTableSchema, String stageTableName,
                                        String tableSchema, String tableName,
                                        TableEntity tableEntity, int stageTableRows) throws SQLException {

        // delete count , insert count
        int[] processCount = {-1, -1};

        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final String columnString = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
        final List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        Map<String, String> stageInfoMap = collectStageTableDataInfo(statement, tableName, stageTableSchema, stageTableName);

        // insert not exist row
        int insertRows = insertNotExistRecordFromStage2TargetTable(statement,
                stageTableSchema, stageTableName, tableSchema, tableName, stageInfoMap, columnString, primaryColumnList, tableEntity);
        processCount[1] = insertRows;

        if (insertRows != stageTableRows) {
            //delete exist
            int deleteRows = deleteTargetTableByStage(statement, stageTableSchema, stageTableName, tableSchema, tableName, stageInfoMap, primaryColumnList);
            processCount[0] = deleteRows;

            //insert all row
            insertRows = insertStage2TargetTable(statement, stageTableSchema, stageTableName, tableSchema, tableName, columnString);
            processCount[1] = insertRows;
        }

        return processCount;
    }

    /**
     * delete + insert
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param tableEntity
     * @return
     * @throws SQLException
     */
    private int[] stageData2TargetTableDL(Statement statement,
                                          String stageTableSchema, String stageTableName,
                                          String tableSchema, String tableName,
                                          TableEntity tableEntity) throws SQLException {

        // delete count , insert count
        int[] processCount = {-1, -1};

        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final String columnString = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
        final List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        Map<String, String> stageInfoMap = collectStageTableDataInfo(statement, tableName, stageTableSchema, stageTableName);

        //delete exist
        int deleteRows = deleteTargetTableByStage(statement, stageTableSchema, stageTableName, tableSchema, tableName, stageInfoMap, primaryColumnList);
        processCount[0] = deleteRows;

        //insert all row
        int insertRows = insertStage2TargetTable(statement, stageTableSchema, stageTableName, tableSchema, tableName, columnString);
        processCount[1] = insertRows;

        return processCount;
    }


    /**
     * insert => catch UNIQUE VIOLATION exception -> delete insert
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param tableEntity
     * @return
     * @throws SQLException
     */
    public int[] stageData2TargetTableFast(Statement statement,
                                            String stageTableSchema, String stageTableName,
                                            String tableSchema, String tableName,
                                            TableEntity tableEntity) throws SQLException {

        // delete count , insert count
        int[] processCount = {-1, -1};

        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final String columnString = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
        final List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        Map<String, String> stageInfoMap = collectStageTableDataInfo(statement, tableName, stageTableSchema, stageTableName);

        int insertRows = 0;
        int deleteRows = -1;
        try {
            //insert all row
            insertRows = insertStage2TargetTable(statement, stageTableSchema, stageTableName, tableSchema, tableName, columnString);
        } catch (SQLException e) {
            if (SQL_STATE_UNIQUE_VIOLATION.equals(e.getSQLState())) {
                deleteRows = deleteTargetTableByStage(statement, stageTableSchema, stageTableName, tableSchema, tableName, stageInfoMap, primaryColumnList);
                processCount[0] = deleteRows;

                insertRows = insertStage2TargetTable(statement, stageTableSchema, stageTableName, tableSchema, tableName, columnString);
                processCount[1] = insertRows;
            } else {
                throw e;
            }
        }
        processCount[0] = deleteRows;
        processCount[1] = insertRows;

        return processCount;
    }

    public int do_insert_records_not_existed(Statement statement, String fromTableSchema, String fromTableName,
                                              String intoTableSchema, String intoTableName, TableEntity tableEntity) throws SQLException {

        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final String columnString = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
        final List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());

        String sql = TableJdbcUtil.buildInsertNotExistedSql(fromTableSchema, fromTableName, intoTableSchema, intoTableName,
                "from_table", "into_table", columnString, primaryColumnList, tableEntity, null);
        logger.info("do insert records not existed , sql: {}", sql);

        final int effectRows = statement.executeUpdate(sql);
        logger.info("insert records not existed effect rows {}", effectRows);
        return effectRows;
    }










    /**
     * 数据从target表插入stage表
     *
     * @param stmt
     * @param tableSchema
     * @param stageTableName
     * @param tableName
     * @param tableEntity
     * @return
     * @throws SQLException
     */
    public int doInsertStageTableFromTarget(Statement stmt, String stageTableName,String tableSchema,String tableName, TableEntity tableEntity) throws SQLException {
        long lBeginTime = System.currentTimeMillis();
        logger.info(" do Insert Stage Table From Target" );

        // over (partition by
        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final List<String> columnNameList = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.toList());
        final String columnString = columnNameList.stream().collect(Collectors.joining(","));
        List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        boolean chamberHis=Arrays.asList((CHAMBER_HST_TABLE).split(",")).stream().filter(element -> stageTableName.contains(element)).count() > 0;
        boolean mdsGlass=Arrays.asList((MDS_GLASS_TABLE).split(",")).stream().filter(element -> stageTableName.contains(element)&&!stageTableName.contains("summary")).count() > 0;

//        if(stageTableName.contains("sht_ope_his")) {
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"logof_timestamp".equals(entity.getColumnName()))
                    .filter(entity -> !"logofTimestamp".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        if(chamberHis){
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }else if(mdsGlass){
                primaryColumnList = primaryColumnList.stream()
                        .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                        .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                        .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                        .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                        .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                        .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                        .filter(entity -> !"equip_id".equals(entity.getColumnName()))
                        .filter(entity -> !"equipId".equals(entity.getColumnName()))
                        .filter(entity -> !"sub_equip_id".equals(entity.getColumnName()))
                        .filter(entity -> !"subEquipId".equals(entity.getColumnName()))
                        .collect(Collectors.toList());
        }



        final String matchString = primaryColumnList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));

        final String fromColumnString = TableJdbcUtil.getColumnStringWithTableAlias(columnEntityList, "from_table");

        // insert
        StringBuilder insertSqlBuilder = new StringBuilder("insert into ");
        insertSqlBuilder.append(stageTableSchema).append(".").append(stageTableName).append("(").append(columnString).append(")");
        insertSqlBuilder.append(" select ").append(fromColumnString).append(" from ")
                .append(" (select * ")
                .append(" from ").append(tableSchema).append(".").append(tableName)
                .append( " where (").append(matchString) .append(") in(").append(" select distinct ").append(matchString).append(" from ")
                .append(stageTableSchema).append(".").append(stageTableName)
                .append(")")
                .append(")").append(" as from_table");

        final String sql = insertSqlBuilder.toString();
        logger.info("targe insert stage sql : {}", sql);
        int effectRows = stmt.executeUpdate(sql);
        logger.info("executeBatch insert {} rows into stage table {} costs {} millis", effectRows, stageTableName, (System.currentTimeMillis() - lBeginTime));

        return effectRows;
    }


    /**
     * update stage表
     *
     * @param stmt
     * @param stageTableName
     * @param tableEntity
     * @return
     * @throws SQLException
     */
    public int updateStageTable(Statement stmt, String stageTableName, TableEntity tableEntity) throws SQLException {

        String timestampColumn="logof_timestamp";
        if(!stageTableName.contains("sht_ope_his")){
            if(stageTableName.contains("array_")||stageTableName.contains("cf_")) timestampColumn="glass_start_time";
            else if(stageTableName.contains("cell_"))timestampColumn="component_start_time";
            else if(stageTableName.contains("cell2_"))timestampColumn="chip_start_time";
        }
        long lBeginTime = System.currentTimeMillis();
        logger.info(" do update StageTable  set   real_flg  = 'N' " );
        StringBuffer sqlBuffer = new StringBuffer("update  ");
        sqlBuffer.append(stageTableSchema).append(".").append(stageTableName);
        sqlBuffer.append(" set  real_flg  = 'N' where 1=1 ");
        logger.info("update stage  set real_flg  = 'N' sql : {}", sqlBuffer.toString());
        int effectRows = stmt.executeUpdate(sqlBuffer.toString());
        logger.info("executeBatch update {} rows into stage table {} costs {} millis", effectRows, stageTableName, (System.currentTimeMillis() - lBeginTime));

        boolean chamberHis=Arrays.asList((CHAMBER_HST_TABLE).split(",")).stream().filter(element -> stageTableName.contains(element)).count() > 0;
        boolean mdsGlass=Arrays.asList((MDS_GLASS_TABLE).split(",")).stream().filter(element -> stageTableName.contains(element)&&!stageTableName.contains("summary")).count() > 0;
        // over (partition by
        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        primaryColumnList = primaryColumnList.stream()
                .filter(entity -> !"logof_timestamp".equals(entity.getColumnName()))
                .filter(entity -> !"logofTimestamp".equals(entity.getColumnName()))
                .collect(Collectors.toList());


        if(chamberHis) {
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }else if(mdsGlass) {
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"equip_id".equals(entity.getColumnName()))
                    .filter(entity -> !"equipId".equals(entity.getColumnName()))
                    .filter(entity -> !"sub_equip_id".equals(entity.getColumnName()))
                    .filter(entity -> !"subEquipId".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }
        final String matchString = primaryColumnList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));

        final String fromColumnString = TableJdbcUtil.getColumnStringWithTableAlias(columnEntityList, "from_table");

        // set  real_flg  = 'Y'
        StringBuilder insertSqlBuilder = new StringBuilder(" update  ").append(stageTableSchema).append(".").append(stageTableName).append(" as stage_table ");
        insertSqlBuilder.append("  set real_flg  = 'Y'  from ");
       /* String partitionOrderCondition = "";
        partitionOrderCondition = " order by logof_timestamp desc ";*/
        insertSqlBuilder.append(" (select ").append(fromColumnString).append(" from ")
                .append(" (select *, row_number() over (partition by ").append(matchString).append(" order by "+timestampColumn+" desc )").append(" as gpload_row_number ")
                .append(" from ").append(stageTableSchema).append(".").append(stageTableName).append(")").append(" as from_table")
                .append(" where gpload_row_number = 1 ) as real_table    where ");

        String updateColumnWhereString = tableEntity.getColumnEntityList().stream()
                .filter(columnEntity -> columnEntity.getPrimaryKeyFlg())
                .map(ColumnEntity::getColumnName)
                .map(column -> String.format(" stage_table.%s = real_table.%s ", column, column))
                .collect(Collectors.joining(" and "));

        insertSqlBuilder.append(updateColumnWhereString);
        final String sql = insertSqlBuilder.toString();
        logger.info("update stage set real_flg  = 'Y' sql : {}", sql);
        effectRows = stmt.executeUpdate(sql);
        logger.info("executeBatch update {} rows into stage table {} costs {} millis", effectRows, stageTableName, (System.currentTimeMillis() - lBeginTime));

        return effectRows;
    }




    /**
     * insert => catch UNIQUE VIOLATION exception -> delete insert
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param tableEntity
     * @return
     * @throws SQLException
     */
    public int[] stageData2TargetTableForOpeHis(Statement statement,
                                                String stageTableSchema, String stageTableName,
                                                String tableSchema, String tableName,
                                                TableEntity tableEntity) throws SQLException {

        // delete count , insert count
        int[] processCount = {-1, -1};

        final List<ColumnEntity> columnEntityList = tableEntity.getColumnEntityList();
        final String columnString = columnEntityList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
        final List<ColumnEntity> primaryColumnList = columnEntityList.stream().filter(ColumnEntity::getPrimaryKeyFlg).collect(Collectors.toList());
        //  Map<String, String> stageInfoMap = collectStageTableDataInfo(statement, tableName, stageTableSchema, stageTableName);

        int insertRows = 0;
        int deleteRows = -1;
        try {
            //insert all row
            insertRows = insertStage2TargetTableForOpeHis(statement, stageTableSchema, stageTableName, tableSchema, tableName, columnString,primaryColumnList);
        } catch (SQLException e) {
            if (SQL_STATE_UNIQUE_VIOLATION.equals(e.getSQLState())) {
                deleteRows = deleteTargetTableByStageForOpeHis(statement, stageTableSchema, stageTableName, tableSchema, tableName, primaryColumnList);
                processCount[0] = deleteRows;

                insertRows = insertStage2TargetTableForOpeHis(statement, stageTableSchema, stageTableName, tableSchema, tableName, columnString,primaryColumnList);
                processCount[1] = insertRows;
            } else {
                throw e;
            }
        }
        processCount[0] = deleteRows;
        processCount[1] = insertRows;

        return processCount;
    }





    /**
     * delete from target table by stage
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param primaryColumnList
     * @return
     * @throws SQLException
     */
    private int deleteTargetTableByStageForOpeHis(Statement statement,
                                                  String stageTableSchema, String stageTableName,
                                                  String tableSchema, String tableName,
                                                  List<ColumnEntity> primaryColumnList) throws SQLException {
        long beginTime = System.currentTimeMillis();

        boolean chamberHis=Arrays.asList((CHAMBER_HST_TABLE).split(",")).stream().filter(element -> stageTableName.contains(element)).count() > 0;
        boolean mdsGlass=Arrays.asList((MDS_GLASS_TABLE).split(",")).stream().filter(element -> stageTableName.contains(element)&&!stageTableName.contains("summary")).count() > 0;

        //delete exist
        primaryColumnList = primaryColumnList.stream()
                .filter(entity -> !"logof_timestamp".equals(entity.getColumnName()))
                .filter(entity -> !"logofTimestamp".equals(entity.getColumnName()))
                .collect(Collectors.toList());


        if(chamberHis){
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }else if(mdsGlass){
            primaryColumnList = primaryColumnList.stream()
                    .filter(entity -> !"glass_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"glassStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"component_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"componentStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"chip_start_time".equals(entity.getColumnName()))
                    .filter(entity -> !"chipStartTime".equals(entity.getColumnName()))
                    .filter(entity -> !"equip_id".equals(entity.getColumnName()))
                    .filter(entity -> !"equipId".equals(entity.getColumnName()))
                    .filter(entity -> !"sub_equip_id".equals(entity.getColumnName()))
                    .filter(entity -> !"subEquipId".equals(entity.getColumnName()))
                    .collect(Collectors.toList());
        }



        final String matchString = primaryColumnList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
        StringBuilder delSql = new StringBuilder(" delete from ");
        delSql.append(tableSchema).append(".").append(tableName)
                .append( " where (").append(matchString) .append(") in(").append(" select distinct ").append(matchString).append(" from ")
                .append(stageTableSchema).append(".").append(stageTableName)
                .append(")");
        String sql = delSql.toString();
        logger.info("do delete by stage, sql: {}", sql);
        final int deleteRows = statement.executeUpdate(sql);
        logger.info("delete total {} rows by stage", deleteRows);
        LogUtil.printPerformance(LOG_MERGE_PROC_DELETE_INSERT_D, tableName, deleteRows, beginTime);
        LogUtil.printCost(LOG_MERGE_PROC_DELETE_INSERT_D, tableName, "delete records rows [" + deleteRows + "]", beginTime);

        return deleteRows;
    }

    /**
     * stage insert into target table
     * 全表插入，调用之前先删除目标表中主键相同的记录
     *
     * @param statement
     * @param stageTableSchema
     * @param stageTableName
     * @param tableSchema
     * @param tableName
     * @param columnString
     * @return
     * @throws SQLException
     */
    public int insertStage2TargetTableForOpeHis(Statement statement,
                                                String stageTableSchema, String stageTableName,
                                                String tableSchema, String tableName,
                                                String columnString,List<ColumnEntity> primaryColumnList) throws SQLException {
        long beginTime = System.currentTimeMillis();
        final String matchString = primaryColumnList.stream().map(ColumnEntity::getColumnName).collect(Collectors.joining(","));
        String sql = new StringBuilder().append("insert into ")
                .append(tableSchema).append(".").append(tableName)
                .append("(").append(columnString).append(")")
                .append("( select distinct ").append(columnString).append(" from ")
//                .append(" (select *, row_number() over (partition by ").append(matchString).append(partitionOrderCondition).append(")").append(" as gpload_row_number ")
                .append(" (select *, row_number() over (partition by ").append(matchString).append(" order by ").append(columnString).append(" desc)").append(" as gpload_row_number ")
                .append(" from ").append(stageTableSchema).append(".").append(stageTableName).append(")").append(" as from_table")
                .append(" where gpload_row_number = 1 )").toString();

        logger.info("do insert from stage, sql: {}.", sql);

        final int insertRows = statement.executeUpdate(sql);
        logger.info("insert {} rows from stage.", insertRows);
        LogUtil.printPerformance(LOG_INSERT_STAGE_2_TARGET_TABLE, tableName, insertRows, beginTime);
        LogUtil.printCost(LOG_INSERT_STAGE_2_TARGET_TABLE, tableName, "insert records rows [" + insertRows + "]", beginTime);

        return insertRows;
    }








}
